package com.shengzai.flink.core

import org.apache.flink.streaming.api.functions.{KeyedProcessFunction, ProcessFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import scala.collection.mutable
import scala.collection.mutable.ListBuffer

object Demo8Fraud {
  /**
   * 对于一个账户，如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易，就输出一个报警信息。
   * 001,20.1
   * 001,30
   * 001,0.07
   * 001,100
   * 001,800
   * 001,20
   * 001,0.1
   * 001,700
   * 001,20
   * 001,0.1
   * 002,700
   * 001,800
   */
  def main(args: Array[String]): Unit = {

      val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

      env.setParallelism(1)
      val hashMap = new mutable.HashMap[String, (Double,Boolean)]()
      //val listBuffer = new ListBuffer[(mutable.HashMap[String, Double], Boolean)]

      val socketDS: DataStream[String] = env.socketTextStream("master", 8888)

//    val mapDS: DataStream[(String, String)] = socketDS.map(line => {
//      val split: Array[String] = line.split(",")
//      (split(0), split(1))
//    })
//
//    val keyS: KeyedStream[(String, String), String] = mapDS.keyBy(_._1)
//
//    keyS.process(new KeyedProcessFunction[String,String,String] {
//      override def processElement(value: String, ctx: KeyedProcessFunction[String, String, String]#Context, out: Collector[String]): Unit = ???
//    })

      socketDS.process(new ProcessFunction[String,mutable.HashMap[String, (Double,Boolean)]] {
        override def processElement(value: String, ctx: ProcessFunction[String, mutable.HashMap[String, (Double, Boolean)]]#Context, out: Collector[mutable.HashMap[String, (Double, Boolean)]]): Unit = {
          val split: Array[String] = value.split(",")
          val id: String = split(0)
          val money: Double = split(1).toDouble
          if (money < 1) {
            hashMap.put(id, (money, true))
          }else if (money > 500 & hashMap.getOrElse(id,(1,false))._2) {
            println("发出警告")
          }else{
            hashMap.put(id, (money, false))
          }

          // hashMap.put(id, (money, false))
          out.collect(hashMap)
        }
      }).print()

    env.execute()
  }
}
